package com.quanyan.user.mq.Consumer;

import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.quanyan.common.constants.Constants;
import com.quanyan.common.utils.ObjectUtil;
import com.quanyan.mq.enums.MqConsumerStatus;
import com.quanyan.mq.mq.consumer.AbstractMqConsumer;
import com.quanyan.mq.mq.consumer.MqTopic;
import com.quanyan.user.mq.UserTaskMsg;
import com.quanyan.user.service.UserTaskService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

/**
 * Created by Administrator on 2016/7/18.
 */
@Component("mqUserTaskActivityConsumer")
public class MqUserTaskActivityConsumer extends AbstractMqConsumer {

    private static final Logger logger = LoggerFactory.getLogger(MqUserTaskConsumer.class);

    private static final String GROUP = "USER_TASK_CONSUMER_GROUP_ACTIVITY";
    private static final String INSTANCE = "USER_TASK_CONSUMER_INSTANCE_ACITYITY";

    @Value("${mq.nameServer.address}")
    private String nameServerAddress;

    @Autowired
    private UserTaskService userTaskService;

    @PostConstruct
    void run() {
        logger.info("用户任务中心mq批量处理消息启动{}",nameServerAddress);
        super.start();
    }

    @Override
    protected MqConsumerStatus process(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
        logger.info("任务中心开始批量活动---------------------------------》接收消息");
        //处理自身业务逻辑
        MessageExt messageExt = messages.get(0);
        if (Constants.MQ_TOPIC_TASK_ACTIVITY_TOPIC.equals(messageExt.getTopic()) && Constants.MQ_TOPIC_TASK_ACTIVITY_TAG.equals(messageExt.getTags())) {
            try {
                logger.debug("处理用户活动任务消息");
                String msgId = messageExt.getMsgId();//mq内部id
                logger.debug("mq消息id:{},业务key :{}", msgId, messageExt.getKeys());
                byte[] bodyByte = messageExt.getBody();
                String msg = ObjectUtil.toObject(bodyByte).toString();
                logger.debug("活动消息是:{}",msg);
                List<UserTaskMsg> userTaskMsg  = (List<UserTaskMsg>) ObjectUtil.toObject(bodyByte);
                if (userTaskMsg != null)
                        userTaskService.doActivityTask(userTaskMsg);
            } catch (Exception e) {
                logger.error("异常信息:{}", e);
                return MqConsumerStatus.RE_CONSUME_LATER;
            }
        }
        return MqConsumerStatus.CONSUMER_SUCCESS;
    }

    @Override
    protected String getConsumerGroup() {
        return this.GROUP;
    }

    @Override
    protected String getNameServerAddress() {
        return this.nameServerAddress;
    }

    @Override
    protected String getInstanceName() {
        return this.INSTANCE;
    }

    @Override
    protected MqTopic[] getMqTopics() {
        MqTopic[] mqTopics = new MqTopic[1];
        mqTopics[0] = new MqTopic(Constants.MQ_TOPIC_TASK_ACTIVITY_TOPIC, Constants.MQ_TOPIC_TASK_ACTIVITY_TAG); //订阅用户做任务的topic与tag
        return mqTopics;
    }
}